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

Tcp Doc Polishing

Browse Source 
Also remove some `this.` method calls.
This commit is contained in:
Gary Russell
2019-02-04 12:32:56 -05:00
committed by Artem Bilan
parent f82c7164f7
commit 6391e4bebf
3 changed files with 70 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2001-2018 the original author or authors.
* Copyright 2001-2019 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.
@@ -81,7 +81,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
*/
@Override
public void close() {
this.setNoReadErrorOnClose(true);
setNoReadErrorOnClose(true);
try {
this.socket.close();
}
@@ -111,8 +111,8 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
this.socketOutputStream.flush();
}
catch (Exception e) {
this.publishConnectionExceptionEvent(new MessagingException(message, "Failed TCP serialization", e));
this.closeConnection(true);
publishConnectionExceptionEvent(new MessagingException(message, "Failed TCP serialization", e));
closeConnection(true);
throw e;
}
if (logger.isDebugEnabled()) {
@@ -122,7 +122,8 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
@Override
public Object getPayload() throws Exception {
return this.getDeserializer().deserialize(inputStream());
return getDeserializer()
.deserialize(inputStream());
}
@Override
@@ -172,16 +173,16 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
public void run() {
boolean okToRun = true;
if (logger.isDebugEnabled()) {
logger.debug(this.getConnectionId() + " Reading...");
logger.debug(getConnectionId() + " Reading...");
}
while (okToRun) {
Message<?> message = null;
try {
message = this.getMapper().toMessage(this);
message = getMapper().toMessage(this);
this.lastRead = System.currentTimeMillis();
}
catch (Exception e) {
this.publishConnectionExceptionEvent(e);
publishConnectionExceptionEvent(e);
if (handleReadException(e)) {
okToRun = false;
}
@@ -218,7 +219,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
* For client connections, we have to wait for 2 timeouts if the last
* send was within the current timeout.
*/
if (!this.isServer() && e instanceof SocketTimeoutException) {
if (!isServer() && e instanceof SocketTimeoutException) {
long now = System.currentTimeMillis();
try {
int soTimeout = this.socket.getSoTimeout();
@@ -226,7 +227,7 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
doClose = false;
}
if (!doClose && logger.isDebugEnabled()) {
logger.debug("Skipping a socket timeout because we have a recent send " + this.getConnectionId());
logger.debug("Skipping a socket timeout because we have a recent send " + getConnectionId());
}
}
catch (SocketException e1) {
@@ -234,39 +235,39 @@ public class TcpNetConnection extends TcpConnectionSupport implements Scheduling
}
}
if (doClose) {
boolean noReadErrorOnClose = this.isNoReadErrorOnClose();
boolean noReadErrorOnClose = isNoReadErrorOnClose();
closeConnection(true);
if (!(e instanceof SoftEndOfStreamException)) {
if (e instanceof SocketTimeoutException) {
if (logger.isDebugEnabled()) {
logger.debug("Closed socket after timeout:" + this.getConnectionId());
logger.debug("Closed socket after timeout:" + getConnectionId());
}
}
else {
if (noReadErrorOnClose) {
if (logger.isTraceEnabled()) {
logger.trace("Read exception " +
this.getConnectionId(), e);
getConnectionId(), e);
}
else if (logger.isDebugEnabled()) {
logger.debug("Read exception " +
this.getConnectionId() + " " +
getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
}
}
else if (logger.isTraceEnabled()) {
logger.error("Read exception " +
this.getConnectionId(), e);
getConnectionId(), e);
}
else {
logger.error("Read exception " +
this.getConnectionId() + " " +
getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + (e.getCause() != null ? e.getCause() + ":" : "") + e.getMessage());
}
}
this.sendExceptionToListener(e);
sendExceptionToListener(e);
}
}
return doClose;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2018 the original author or authors.
* Copyright 2002-2019 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.
@@ -115,7 +115,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
@Override
public void close() {
this.setNoReadErrorOnClose(true);
setNoReadErrorOnClose(true);
doClose();
}
@@ -144,19 +144,19 @@ public class TcpNioConnection extends TcpConnectionSupport {
synchronized (this.socketChannel) {
if (this.bufferedOutputStream == null) {
int writeBufferSize = this.socketChannel.socket().getSendBufferSize();
this.bufferedOutputStream = new BufferedOutputStream(this.getChannelOutputStream(),
this.bufferedOutputStream = new BufferedOutputStream(getChannelOutputStream(),
writeBufferSize > 0 ? writeBufferSize : 8192);
}
Object object = this.getMapper().fromMessage(message);
Object object = getMapper().fromMessage(message);
Assert.state(object != null, "Mapper mapped the message to 'null'.");
this.lastSend = System.currentTimeMillis();
try {
((Serializer<Object>) this.getSerializer()).serialize(object, this.bufferedOutputStream);
((Serializer<Object>) getSerializer()).serialize(object, this.bufferedOutputStream);
this.bufferedOutputStream.flush();
}
catch (Exception e) {
this.publishConnectionExceptionEvent(new MessagingException(message, "Failed TCP serialization", e));
this.closeConnection(true);
publishConnectionExceptionEvent(new MessagingException(message, "Failed TCP serialization", e));
closeConnection(true);
throw e;
}
if (logger.isDebugEnabled()) {
@@ -167,7 +167,8 @@ public class TcpNioConnection extends TcpConnectionSupport {
@Override
public Object getPayload() throws Exception {
return this.getDeserializer().deserialize(inputStream());
return getDeserializer()
.deserialize(inputStream());
}
@Override
@@ -224,7 +225,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
@Override
public void run() {
if (logger.isTraceEnabled()) {
logger.trace(this.getConnectionId() + " Nio message assembler running...");
logger.trace(getConnectionId() + " Nio message assembler running...");
}
boolean moreDataAvailable = true;
while (moreDataAvailable) {
@@ -242,8 +243,10 @@ public class TcpNioConnection extends TcpConnectionSupport {
catch (RejectedExecutionException e) {
this.executionControl.decrementAndGet();
if (logger.isInfoEnabled()) {
logger.info(getConnectionId() + " Insufficient threads in the assembler fixed thread pool; consider " +
"increasing this task executor pool size; data avail: " + this.channelInputStream.available());
logger.info(getConnectionId()
+ " Insufficient threads in the assembler fixed thread pool; consider "
+ "increasing this task executor pool size; data avail: "
+ this.channelInputStream.available());
}
}
}
@@ -259,24 +262,24 @@ public class TcpNioConnection extends TcpConnectionSupport {
catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.error("Read exception " +
this.getConnectionId(), e);
getConnectionId(), e);
}
else if (!this.isNoReadErrorOnClose()) {
else if (!isNoReadErrorOnClose()) {
logger.error("Read exception " +
this.getConnectionId() + " " +
getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Read exception " +
this.getConnectionId() + " " +
getConnectionId() + " " +
e.getClass().getSimpleName() +
":" + e.getCause() + ":" + e.getMessage());
}
}
this.closeConnection(true);
this.sendExceptionToListener(e);
closeConnection(true);
sendExceptionToListener(e);
return;
}
}
@@ -301,12 +304,13 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
if (moreDataAvailable) {
if (logger.isTraceEnabled()) {
logger.trace(this.getConnectionId() + " Nio message assembler continuing...");
logger.trace(getConnectionId() + " Nio message assembler continuing...");
}
}
else {
if (logger.isTraceEnabled()) {
logger.trace(this.getConnectionId() + " Nio message assembler exiting... avail: " + this.channelInputStream.available());
logger.trace(getConnectionId() + " Nio message assembler exiting... avail: "
+ this.channelInputStream.available());
}
}
}
@@ -355,13 +359,13 @@ public class TcpNioConnection extends TcpConnectionSupport {
}
Message<?> message = null;
try {
message = this.getMapper().toMessage(this);
message = getMapper().toMessage(this);
}
catch (Exception e) {
this.closeConnection(true);
closeConnection(true);
if (e instanceof SocketTimeoutException) {
if (logger.isDebugEnabled()) {
logger.debug("Closing socket after timeout " + this.getConnectionId());
logger.debug("Closing socket after timeout " + getConnectionId());
}
}
else {
@@ -420,7 +424,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
int len = this.socketChannel.read(this.rawBuffer);
if (len < 0) {
this.writingToPipe = false;
this.closeConnection(true);
closeConnection(true);
}
if (logger.isTraceEnabled()) {
logger.trace("After read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit());
@@ -432,13 +436,13 @@ public class TcpNioConnection extends TcpConnectionSupport {
if (logger.isDebugEnabled()) {
logger.debug("Read " + this.rawBuffer.limit() + " into raw buffer");
}
this.sendToPipe(this.rawBuffer);
sendToPipe(this.rawBuffer);
}
catch (RejectedExecutionException e) {
throw e;
}
catch (Exception e) {
this.publishConnectionExceptionEvent(e);
publishConnectionExceptionEvent(e);
throw e;
}
finally {
@@ -450,7 +454,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
protected void sendToPipe(ByteBuffer rawBuffer) throws IOException {
Assert.notNull(rawBuffer, "rawBuffer cannot be null");
if (logger.isTraceEnabled()) {
logger.trace(this.getConnectionId() + " Sending " + rawBuffer.limit() + " to pipe");
logger.trace(getConnectionId() + " Sending " + rawBuffer.limit() + " to pipe");
}
this.channelInputStream.write(rawBuffer);
rawBuffer.clear();
@@ -462,7 +466,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
// only execute run() if we don't already have one running
this.executionControl.set(1);
if (logger.isDebugEnabled()) {
logger.debug(this.getConnectionId() + " Running an assembler");
logger.debug(getConnectionId() + " Running an assembler");
}
try {
this.taskExecutor.execute2(this);
@@ -487,25 +491,25 @@ public class TcpNioConnection extends TcpConnectionSupport {
*/
public void readPacket() {
if (logger.isDebugEnabled()) {
logger.debug(this.getConnectionId() + " Reading...");
logger.debug(getConnectionId() + " Reading...");
}
try {
doRead();
}
catch (ClosedChannelException cce) {
if (logger.isDebugEnabled()) {
logger.debug(this.getConnectionId() + " Channel is closed");
logger.debug(getConnectionId() + " Channel is closed");
}
this.closeConnection(true);
closeConnection(true);
}
catch (RejectedExecutionException e) {
throw e;
}
catch (Exception e) {
logger.error("Exception on Read " +
this.getConnectionId() + " " +
getConnectionId() + " " +
e.getMessage(), e);
this.closeConnection(true);
closeConnection(true);
}
}
@@ -514,7 +518,7 @@ public class TcpNioConnection extends TcpConnectionSupport {
*/
void timeout() {
this.timedOut = true;
this.closeConnection(true);
closeConnection(true);
}
/**

17
src/reference/asciidoc/ip.adoc
View File

@@ -291,6 +291,8 @@ public IntegrationFlow udpEchoUpcaseServer() {
[[tcp-connection-factories]]
=== TCP Connection Factories
==== Overview
For TCP, the configuration of the underlying connection is provided by using a connection factory.
Two types of connection factory are provided: a client connection factory and a server connection factory.
Client connection factories establish outgoing connections.
@@ -371,6 +373,8 @@ The following example shows a client connection factory that uses `java.net.Sock
----
====
==== Message Demarcation (Serializers and Deserializers)
TCP is a streaming protocol.
This means that some structure has to be provided to data transported over TCP so that the receiver can demarcate the data into discrete messages.
Connection factories are configured to use serializers and deserializers to convert between the message payload and the bits that are sent over TCP.
@@ -437,6 +441,8 @@ This is similar to the deserializer side of `ByteArrayRawSerializer` above, exce
Internally, it uses a `ByteArrayOutputStream` that lets the buffer grow as needed.
The client must close the socket in an orderly manner to signal end of message.
WARNING: This deserializer should only be used when the peer is trusted; it is susceptible to a DoS attach due to out of memory conditions.
The `MapJsonSerializer` uses a Jackson `ObjectMapper` to convert between a `Map` and JSON.
You can use this serializer in conjunction with a `MessageConvertingTcpMessageMapper` and a `MapMessageConverter` to transfer selected headers and the payload in JSON.
@@ -447,8 +453,6 @@ By default, a `ByteArrayLfSerializer` is used, resulting in messages with a form
The final standard serializer is `org.springframework.core.serializer.DefaultSerializer`, which you can use to convert serializable objects with Java serialization.
`org.springframework.core.serializer.DefaultDeserializer` is provided for inbound deserialization of streams that contain serializable objects.
To implement a custom serializer and deserializer pair, implement the `org.springframework.core.serializer.Deserializer` and `org.springframework.core.serializer.Serializer` interfaces.
If you do not wish to use the default serializer and deserializer (`ByteArrayCrLfSerializer`), you must set the `serializer` and `deserializer` attributes on the connection factory.
The following example shows how to do so:
@@ -480,6 +484,15 @@ NOTE: You can also modify the attributes of sockets and socket factories.
See <<ssl-tls>>.
As noted there, such modifications are possible whether or not SSL is being used.
==== Custom Serializers and Deserializers
If your data is not in a format supported by one of the standard deserializers, you can implement your own; you can also implement a custom serializer.
To implement a custom serializer and deserializer pair, implement the `org.springframework.core.serializer.Deserializer` and `org.springframework.core.serializer.Serializer` interfaces.
When the deserializer detects a closed input stream between messages, it must throw a `SoftEndOfStreamException`; this is a signal to the framework to indicate that the close was "normal".
If the stream is closed while decoding a message, some other exception should be thrown instead.
[[caching-cf]]
==== TCP Caching Client Connection Factory