Skip to content

Commit 97e2d10

Browse files
petrov-mgpetrov-mg
authored
IGNITE-29001 Removed unused Message parameter for TcpDiscoverySpi#writeToSocket methods (#13505)
1 parent 427cb85 commit 97e2d10

19 files changed

Lines changed: 169 additions & 126 deletions

‎modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java‎

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -6861,7 +6861,7 @@ else if (e.hasCause(ObjectStreamException.class) || (!sock.isClosed() && !e.hasC
68616861
if (msg instanceof TcpDiscoveryConnectionCheckMessage) {
68626862
ringMessageReceived();
68636863

6864-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
6864+
spi.writeToSocket(sock, RES_OK, sockTimeout);
68656865

68666866
continue;
68676867
}
@@ -6887,7 +6887,7 @@ else if (msg instanceof TcpDiscoveryClientReconnectMessage) {
68876887
TcpDiscoverySpiState state = spiStateCopy();
68886888

68896889
if (state == CONNECTED) {
6890-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
6890+
spi.writeToSocket(sock, RES_OK, sockTimeout);
68916891

68926892
if (clientMsgWrk != null && clientMsgWrk.runner() == null && !clientMsgWrk.isDone())
68936893
new MessageWorkerThreadWithCleanup<>(clientMsgWrk, log).start();
@@ -6901,21 +6901,21 @@ else if (msg instanceof TcpDiscoveryClientReconnectMessage) {
69016901

69026902
// If message is received from previous node and node is connecting forward to next node.
69036903
if (!getLocalNodeId().equals(msg0.routerNodeId()) && state == CONNECTING) {
6904-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
6904+
spi.writeToSocket(sock, RES_OK, sockTimeout);
69056905

69066906
msgWorker.addMessage(msg);
69076907

69086908
continue;
69096909
}
69106910

6911-
spi.writeToSocket(msg, sock, RES_CONTINUE_JOIN, sockTimeout);
6911+
spi.writeToSocket(sock, RES_CONTINUE_JOIN, sockTimeout);
69126912

69136913
break;
69146914
}
69156915
}
69166916
else if (msg instanceof TcpDiscoveryDuplicateIdMessage) {
69176917
// Send receipt back.
6918-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
6918+
spi.writeToSocket(sock, RES_OK, sockTimeout);
69196919

69206920
boolean ignored = false;
69216921

@@ -6944,7 +6944,7 @@ else if (msg instanceof TcpDiscoveryDuplicateIdMessage) {
69446944
}
69456945
else if (msg instanceof TcpDiscoveryAuthFailedMessage) {
69466946
// Send receipt back.
6947-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
6947+
spi.writeToSocket(sock, RES_OK, sockTimeout);
69486948

69496949
synchronized (mux) {
69506950
if (spiState == CONNECTING) {
@@ -6972,7 +6972,7 @@ else if (msg instanceof TcpDiscoveryAuthFailedMessage) {
69726972
}
69736973
else if (msg instanceof TcpDiscoveryCheckFailedMessage) {
69746974
// Send receipt back.
6975-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
6975+
spi.writeToSocket(sock, RES_OK, sockTimeout);
69766976

69776977
boolean ignored = false;
69786978

@@ -7015,7 +7015,7 @@ else if (msg instanceof TcpDiscoveryCheckFailedMessage) {
70157015
}
70167016
else if (msg instanceof TcpDiscoveryLoopbackProblemMessage) {
70177017
// Send receipt back.
7018-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
7018+
spi.writeToSocket(sock, RES_OK, sockTimeout);
70197019

70207020
boolean ignored = false;
70217021

@@ -7080,7 +7080,7 @@ else if (msg instanceof TcpDiscoveryRingLatencyCheckMessage) {
70807080
clientMsgWrk.addMessage(ack);
70817081
}
70827082
else
7083-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
7083+
spi.writeToSocket(sock, RES_OK, sockTimeout);
70847084

70857085
if (metricsUpdateMsg != null)
70867086
processClientMetricsUpdateMessage(metricsUpdateMsg);
@@ -7398,13 +7398,13 @@ private boolean processJoinRequestMessage(
73987398
// Check that joining node can accept incoming connections.
73997399
if (node.clientRouterNodeId() == null) {
74007400
if (!pingJoiningNode(node)) {
7401-
spi.writeToSocket(msg, sock, RES_JOIN_IMPOSSIBLE, sockTimeout);
7401+
spi.writeToSocket(sock, RES_JOIN_IMPOSSIBLE, sockTimeout);
74027402

74037403
return false;
74047404
}
74057405
}
74067406

7407-
spi.writeToSocket(msg, sock, RES_OK, sockTimeout);
7407+
spi.writeToSocket(sock, RES_OK, sockTimeout);
74087408

74097409
if (log.isDebugEnabled())
74107410
log.debug("Responded to join request message [msg=" + msg + ", res=" + RES_OK + ']');
@@ -7441,7 +7441,7 @@ private boolean processJoinRequestMessage(
74417441
// Local node is stopping. Remote node should try next one.
74427442
res = RES_CONTINUE_JOIN;
74437443

7444-
spi.writeToSocket(msg, sock, res, sockTimeout);
7444+
spi.writeToSocket(sock, res, sockTimeout);
74457445

74467446
if (log.isDebugEnabled())
74477447
log.debug("Responded to join request message [msg=" + msg + ", res=" + res + ']');
@@ -7719,7 +7719,7 @@ private void writeToSocket(T2<TcpDiscoveryAbstractMessage, byte[]> msgT, long ti
77197719
throws IgniteCheckedException, IOException {
77207720
byte[] msgBytes = msgT.get2() == null ? clientMsgSer.serializeMessage(msgT.get1()) : msgT.get2();
77217721

7722-
spi.writeToSocket(sock, msgT.get1(), msgBytes, timeout);
7722+
spi.writeToSocket(sock, msgBytes, timeout);
77237723
}
77247724

77257725
/**

‎modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1617,7 +1617,7 @@ protected Socket openSocket(
16171617

16181618
sock.connect(resolved, (int)timeoutHelper.nextTimeoutChunk(sockTimeout));
16191619

1620-
writeToSocket(sock, null, U.IGNITE_HEADER, timeoutHelper.nextTimeoutChunk(sockTimeout));
1620+
writeToSocket(sock, U.IGNITE_HEADER, timeoutHelper.nextTimeoutChunk(sockTimeout));
16211621

16221622
return sock;
16231623
}
@@ -1722,18 +1722,16 @@ void validateRemoteFeatures(IgniteNodeFeatureSet rmtFeatures) throws IgniteCheck
17221722
}
17231723

17241724
/**
1725-
* Writes message to the socket.
1725+
* Writes raw data to the socket.
17261726
*
17271727
* @param sock Socket.
1728-
* @param msg Message.
17291728
* @param data Raw data to write.
17301729
* @param timeout Socket write timeout.
17311730
* @throws IOException If IO failed or write timed out.
17321731
* @throws IgniteCheckedException If node is not yet initialized or is stopping.
17331732
*/
17341733
protected void writeToSocket(
17351734
Socket sock,
1736-
@Nullable TcpDiscoveryAbstractMessage msg,
17371735
byte[] data,
17381736
long timeout
17391737
) throws IOException, IgniteCheckedException {
@@ -1808,15 +1806,13 @@ protected void writeMessage(
18081806
/**
18091807
* Writes response to the socket.
18101808
*
1811-
* @param msg Received message.
18121809
* @param sock Socket.
18131810
* @param res Integer response.
18141811
* @param timeout Socket timeout.
18151812
* @throws IOException If IO failed or write timed out.
18161813
* @throws IgniteCheckedException If node is not yet initialized or is stopping.
18171814
*/
18181815
protected void writeToSocket(
1819-
TcpDiscoveryAbstractMessage msg,
18201816
Socket sock,
18211817
int res,
18221818
long timeout

‎modules/core/src/test/java/org/apache/ignite/internal/IgniteClientRejoinTest.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -355,14 +355,13 @@ private class DiscoverySpi extends TcpDiscoverySpi {
355355
/** {@inheritDoc} */
356356
@Override protected void writeToSocket(
357357
Socket sock,
358-
TcpDiscoveryAbstractMessage msg,
359358
byte[] data,
360359
long timeout
361360
) throws IOException, IgniteCheckedException {
362361
if (blockAll || block && sock.getPort() == 47500)
363362
throw new SocketException("Test discovery exception");
364363

365-
super.writeToSocket(sock, msg, data, timeout);
364+
super.writeToSocket(sock, data, timeout);
366365
}
367366

368367
/** {@inheritDoc} */
@@ -379,15 +378,14 @@ private class DiscoverySpi extends TcpDiscoverySpi {
379378

380379
/** {@inheritDoc} */
381380
@Override protected void writeToSocket(
382-
TcpDiscoveryAbstractMessage msg,
383381
Socket sock,
384382
int res,
385383
long timeout
386384
) throws IOException, IgniteCheckedException {
387385
if (blockAll || block && sock.getPort() == 47500)
388386
throw new SocketException("Test discovery exception");
389387

390-
super.writeToSocket(msg, sock, res, timeout);
388+
super.writeToSocket(sock, res, timeout);
391389
}
392390

393391
/** {@inheritDoc} */

‎modules/core/src/test/java/org/apache/ignite/internal/IgniteDiscoveryMassiveNodeFailTest.java‎

Lines changed: 11 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -304,54 +304,48 @@ private class FailDiscoverySpi extends TcpDiscoverySpi {
304304
/** {@inheritDoc} */
305305
@Override protected void writeToSocket(
306306
Socket sock,
307-
TcpDiscoveryAbstractMessage msg,
308307
byte[] data,
309308
long timeout
310309
) throws IOException, IgniteCheckedException {
311310
assertNotFailedNode(sock);
312311

313-
if (isDrop(msg))
312+
if (isDrop())
314313
return;
315314

316-
super.writeToSocket(sock, msg, data, timeout);
315+
super.writeToSocket(sock, data, timeout);
317316
}
318317

319318
/** {@inheritDoc} */
320319
@Override protected void writeMessage(TcpDiscoveryIoSession ses, TcpDiscoveryAbstractMessage msg,
321320
long timeout) throws IOException, IgniteCheckedException {
322321
assertNotFailedNode(ses.socket());
323322

324-
if (isDrop(msg))
323+
if (isDrop()) {
324+
ignite.log().info(">> Drop message " + msg);
325+
325326
return;
327+
}
326328

327329
super.writeMessage(ses, msg, timeout);
328330
}
329331

330-
/**
331-
*
332-
*/
333-
private boolean isDrop(TcpDiscoveryAbstractMessage msg) {
334-
boolean drop = failNodes && forceFailConnectivity && failedNodes.contains(ignite.cluster().localNode());
335-
336-
if (drop)
337-
ignite.log().info(">> Drop message " + msg);
338-
339-
return drop;
332+
/** */
333+
private boolean isDrop() {
334+
return failNodes && forceFailConnectivity && failedNodes.contains(ignite.cluster().localNode());
340335
}
341336

342337
/** {@inheritDoc} */
343338
@Override protected void writeToSocket(
344-
TcpDiscoveryAbstractMessage msg,
345339
Socket sock,
346340
int res,
347341
long timeout
348342
) throws IOException, IgniteCheckedException {
349343
assertNotFailedNode(sock);
350344

351-
if (isDrop(msg))
345+
if (isDrop())
352346
return;
353347

354-
super.writeToSocket(msg, sock, res, timeout);
348+
super.writeToSocket(sock, res, timeout);
355349
}
356350

357351
/**

‎modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTopologySplitAbstractTest.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -220,13 +220,12 @@ protected void checkSegmented(InetSocketAddress sockAddr, long timeout) throws S
220220
/** {@inheritDoc} */
221221
@Override protected void writeToSocket(
222222
Socket sock,
223-
TcpDiscoveryAbstractMessage msg,
224223
byte[] data,
225224
long timeout
226225
) throws IOException, IgniteCheckedException {
227226
checkSegmented((InetSocketAddress)sock.getRemoteSocketAddress(), timeout);
228227

229-
super.writeToSocket(sock, msg, data, timeout);
228+
super.writeToSocket(sock, data, timeout);
230229
}
231230

232231
/** {@inheritDoc} */
@@ -240,14 +239,13 @@ protected void checkSegmented(InetSocketAddress sockAddr, long timeout) throws S
240239

241240
/** {@inheritDoc} */
242241
@Override protected void writeToSocket(
243-
TcpDiscoveryAbstractMessage msg,
244242
Socket sock,
245243
int res,
246244
long timeout
247245
) throws IOException, IgniteCheckedException {
248246
checkSegmented((InetSocketAddress)sock.getRemoteSocketAddress(), timeout);
249247

250-
super.writeToSocket(msg, sock, res, timeout);
248+
super.writeToSocket(sock, res, timeout);
251249
}
252250
}
253251

‎modules/core/src/test/java/org/apache/ignite/internal/processors/rest/RestProcessorHangTest.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@
3131
import org.apache.ignite.internal.IgnitionEx;
3232
import org.apache.ignite.internal.processors.rest.request.GridRestCacheRequest;
3333
import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
34-
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
3534
import org.apache.ignite.testframework.GridTestUtils;
3635
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
3736
import org.junit.Test;
@@ -73,7 +72,6 @@ public void testNodeStopOnDiscoverySpiFailTest() throws Exception {
7372
// Discovery spi that never allows connecting.
7473
TestTcpDiscoverySpi discoSpi = new TestTcpDiscoverySpi() {
7574
@Override protected void writeToSocket(
76-
TcpDiscoveryAbstractMessage msg,
7775
Socket sock,
7876
int res,
7977
long timeout
@@ -86,7 +84,7 @@ public void testNodeStopOnDiscoverySpiFailTest() throws Exception {
8684
// No-op.
8785
}
8886

89-
super.writeToSocket(msg, sock, 255, timeout);
87+
super.writeToSocket(sock, 255, timeout);
9088
}
9189
};
9290

‎modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationConnectOnInitTest.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -202,13 +202,12 @@ private class TestDiscoverySpi extends TcpDiscoverySpi {
202202
/** {@inheritDoc} */
203203
@Override protected void writeToSocket(
204204
Socket sock,
205-
TcpDiscoveryAbstractMessage msg,
206205
byte[] data,
207206
long timeout
208207
) throws IOException, IgniteCheckedException {
209208
awaitLatch();
210209

211-
super.writeToSocket(sock, msg, data, timeout);
210+
super.writeToSocket(sock, data, timeout);
212211
}
213212

214213
/** {@inheritDoc} */
@@ -224,14 +223,13 @@ private class TestDiscoverySpi extends TcpDiscoverySpi {
224223

225224
/** {@inheritDoc} */
226225
@Override protected void writeToSocket(
227-
TcpDiscoveryAbstractMessage msg,
228226
Socket sock,
229227
int res,
230228
long timeout
231229
) throws IOException, IgniteCheckedException {
232230
awaitLatch();
233231

234-
super.writeToSocket(msg, sock, res, timeout);
232+
super.writeToSocket(sock, res, timeout);
235233
}
236234

237235
/**

‎modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/BlockTcpDiscoverySpi.java‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -67,14 +67,17 @@ private synchronized void apply(ClusterNode addr, TcpDiscoveryAbstractMessage ms
6767
/** {@inheritDoc} */
6868
@Override protected void writeToSocket(
6969
Socket sock,
70-
TcpDiscoveryAbstractMessage msg,
7170
byte[] data,
7271
long timeout
7372
) throws IOException, IgniteCheckedException {
74-
if (spiCtx != null)
75-
apply(spiCtx.localNode(), msg);
73+
if (spiCtx != null) {
74+
TcpDiscoveryAbstractMessage msg = decodeMessage(this, data);
75+
76+
if (msg != null)
77+
apply(spiCtx.localNode(), msg);
78+
}
7679

77-
super.writeToSocket(sock, msg, data, timeout);
80+
super.writeToSocket(sock, data, timeout);
7881
}
7982

8083
/** {@inheritDoc} */

‎modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -351,11 +351,10 @@ private TestTcpDiscoverySpi(
351351
}
352352

353353
/** {@inheritDoc} */
354-
@Override protected void writeToSocket(Socket sock, @Nullable TcpDiscoveryAbstractMessage msg, byte[] data,
355-
long timeout) throws IOException, IgniteCheckedException {
354+
@Override protected void writeToSocket(Socket sock, byte[] data, long timeout) throws IOException, IgniteCheckedException {
356355
tryToBlock(sock, data, timeout);
357356

358-
super.writeToSocket(sock, msg, data, timeout);
357+
super.writeToSocket(sock, data, timeout);
359358
}
360359

361360
/** */

0 commit comments

Comments
 (0)